package com.platform.apex.app;

import org.apache.hadoop.conf.Configuration;

import com.datatorrent.api.DAG;
import com.datatorrent.api.DAG.Locality;
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.annotation.ApplicationAnnotation;
import com.platform.apex.input.HDFSInoutOperator;
import com.platform.apex.output.HDFSOutputOperator;
import com.platform.apex.transform.HDFSToJsonOperatorHgt;

@ApplicationAnnotation(name = "HDFSToHDFSApplicaton")
public class HDFSToHDFSApplicaton implements StreamingApplication{

	@Override
	public void populateDAG(DAG dag, Configuration conf) {
		loadDag(dag, conf);
		
	}

	
	private void loadDag(DAG dag, Configuration conf) {
//		MergeOp wd = dag.addOperator("mg", MergeOp.class);
//		HDFSInoutOperator hi = dag.addOperator("hi", HDFSInoutOperator.class);
		HDFSToJsonOperatorHgt htgout = dag.addOperator("htgout", HDFSToJsonOperatorHgt.class);
		
//	    dag.getOperatorMeta("htgout").getAttributes().put(OperatorContext.APPLICATION_WINDOW_COUNT, 10);
//		HDFSToJsonOperatorUwnd uwndout = dag.addOperator("uwndout", HDFSToJsonOperatorUwnd.class);
		HDFSOutputOperator ho = dag.addOperator("ho", HDFSOutputOperator.class);
		
		
		
//		ConsoleOutputOperator cons = dag.addOperator("console", new ConsoleOutputOperator());
		
		
//		dag.addStream("hi-to-ht-hgt", hi.outhgt, htgout.inputPorthgt).setLocality(Locality.CONTAINER_LOCAL);
		dag.addStream("ht-to-ho-hgt", htgout.outhgt, ho.inputPort);
//		dag.addStream("hi-to-ht-uwnd", hi.outuwndandvwnd, htgout.inputPortuwndandvwnd);
		
//		dag.addStream("ht-to-con", ht.out, cons.input);
//		dag.addStream("ht-to-ho", ht.out, ho.inputPort);
	}
}
